package client

import (
	"errors"
	"io"
	"log"
	"net"
	"proxyRelayServer/src/common"
	"proxyRelayServer/src/util"
	"sync"
	"sync/atomic"
	"time"
)

type Subscriber struct {
	//Name               string
	subAddr            net.Addr
	TcpConnCoreSize    int
	serviceAddr        net.Addr
	subConnSizeCounter atomic.Int32
	lock               *sync.Mutex
	cond               *sync.Cond
}

func NewSubscriber(
	subAddr net.Addr,
	serviceAddr net.Addr,
	tcpConnCoreSize int) *Subscriber {
	l := &sync.Mutex{}
	cond := sync.NewCond(l)
	return &Subscriber{
		subAddr:         subAddr,
		serviceAddr:     serviceAddr,
		TcpConnCoreSize: tcpConnCoreSize,
		lock:            l,
		cond:            cond,
	}
}

func (s *Subscriber) Start() {
	for {
		log.Println("sub")
		subscribe := s.subscribe()
		if subscribe == nil {
			time.Sleep(1 * time.Second)
			continue
		}
		log.Println("create sub conn")
		go func() {
			defer func() {
				if err := recover(); err != nil {
					log.Println(err)
					time.Sleep(200 * time.Millisecond)
				}
				//异步函数结束 说明中继连接已经释放 需要重新建立
				s.subConnSizeCounter.Add(-1)
				s.cond.Signal()
				_ = subscribe.Close()
			}()
			connection := common.BuildConnection(subscribe)
			//阻塞获取中继服务器输入 可闲置十分钟
			err := connection.SetReadDeadline(time.Now().Add(time.Second * 600))
			if err != nil {
				log.Println(err)
				return
			}
			//阻塞至远端写入数据
			if !connection.RxData() {
				time.Sleep(time.Second)
				return
			}
			log.Println("conn rx data")
		reConn:
			serviceConn, err := s.netDial()
			if err != nil {
				log.Println(err)
				time.Sleep(5 * time.Second)
				goto reConn
			}
			defer func() {
				_ = serviceConn.Close()
			}()
			group := sync.WaitGroup{}
			group.Add(1)
			go func() {
				defer group.Done()
				written, err2 := util.CopyBuf(connection, serviceConn, time.Second*30, time.Second*30)
				if err2 != nil {
					log.Println(err2)
					err2 := connection.Close()
					if err2 != nil {
						log.Println(err2)
					}
				}
				log.Printf("service -> %s %d bytes", subscribe.RemoteAddr(), written)
			}()
			written, err := util.CopyBuf(serviceConn, connection, time.Second*30, time.Second*30)
			if err != nil {
				if !errors.Is(err, io.EOF) {
					log.Println(err)
					err := serviceConn.Close()
					if err != nil {
						log.Println(err)
					}
				}
			}
			log.Printf("%s -> service %d bytes", connection.RemoteAddr(), written)
			//等待异步读写结束
			group.Wait()
		}()
	}
}

func (s *Subscriber) netDial() (net.Conn, error) {
	conn, err := net.Dial("tcp", s.serviceAddr.String())
	return conn, err
}

func (s *Subscriber) subscribe() net.Conn {
	s.lock.Lock()
	defer s.lock.Unlock()
	for int(s.subConnSizeCounter.Load()) >= s.TcpConnCoreSize {
		s.cond.Wait()
	}
	dial, err := net.Dial("tcp", s.subAddr.String())
	if err != nil {
		log.Println(err)
		return nil
	}

	s.subConnSizeCounter.Add(1)
	return dial
}
